package cn.itcast.zt.demo01;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

/**
 * 主函数进行测试
 * 每隔一秒事件源sendData(),消费者就开始消费数据，这里实质上是一种观察者模式，内部状态改变会通知所有的消费者
 * 参考：
 *  https://my.oschina.net/manmao/blog/803080
 * create by zhangtian3 on 2018/10/9
 */
public class StringEventMain {
    public static void main(String[] args) throws Exception{
        //创建一个线程池
        // ExecutorService executorPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        // 创建Event工厂
        StringEventFactory factory = new StringEventFactory();
        /**创建Disruptor对象
         * 创建Disruptor对象
         * eventFactory, 传入实现了EventFactory接口的工厂类
         * ringBufferSize, 用来存储数据的， 值为 2^n
         * executor, 线程池
         * producerType, 类型，可以是多个生产者，也可以是单个生产者
         * waitStrategy, 使用什么策略，消费者如何等待生产者放入disruptor中:
         *      BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现
         *      SleepingWaitStrategy 的性能表现跟BlockingWaitStrategy差不多，对CPU的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景
         *      YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于CPU逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性
         */
        // 该构造器已过期，替换了ThreadFactory创建线程
        // Disruptor<StringEvent> disruptor = new Disruptor<StringEvent>(factory, (int)Math.pow(2, 20), executorPool, ProducerType.SINGLE, new YieldingWaitStrategy());

        // ThreadFactory 代替 ExecutorService
        ThreadFactory tf = new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r, "Product-Thread") ;
                t.setDaemon(true);
                return t;
            }
        } ;

        Disruptor<StringEvent> disruptor = new Disruptor<StringEvent>(factory, (int)Math.pow(2, 20), tf, ProducerType.SINGLE, new YieldingWaitStrategy());
        // 关联处理器，也就是消费者，连接消费事件方法
        disruptor.handleEventsWith(new StringEventHandler());

        // ----------------------- 场景模拟增强 start ------------------------
        // 类型1 多个消费者每个消费者都有机会消费相同数据，使用handleEventsWith方法
        // disruptor.handleEventsWith(new ComsumerHandler(1),new ComsumerHandler(2)) ;
        // 类型2 多个消费者，每个消费者消费不同数据。也就是说每个消费者竞争数据，竞争到消费，其他消费者没有机会。使用handleEventsWithWorkerPool方法
        // disruptor.handleEventsWithWorkerPool(new ComsumerHandler(1),new ComsumerHandler(2)) ;
        // ----------------------- 场景模拟增强 end -------------------------

        // 启动
        disruptor.start();

        // 获取RingBuffer,模拟生产者发布消息
        RingBuffer<StringEvent> ringBuffer = disruptor.getRingBuffer();
        // disruptor3推荐的写法
        StringEventProducerWithTranslator producer = new StringEventProducerWithTranslator(ringBuffer);
        // StringEventProducer producer = new StringEventProducer(ringBuffer);
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        // 闭锁控制线程同步
        CountDownLatch countDownLatch = new CountDownLatch(1);

        new Thread(new Runnable() {
            @Override
            public void run() {
                for(int i = 0; i < 10 ; i ++) {
                    // 下面是进行触发事件并且发布
                    byteBuffer.put(new String("生产者"+ i +"发布消息").getBytes());
                    producer.sendData(byteBuffer);
                    // 模拟进行其他操作的耗时
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                countDownLatch.countDown();
            }
        },"Thread2").start();;

        // 等待
        countDownLatch.await();
        disruptor.shutdown();// 关闭 disruptor
        //executorPool.shutdown();// 关闭线程池
    }
}
